package com.zk.storm;
//每日浏览器总点击量
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

public class DailyAgentSumBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Map<String, Long> agentSumMap;  // 键：date#agent，值：总点击量

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.agentSumMap = new HashMap<>();
    }

    @Override
    public void execute(Tuple input) {
        try {
            //提取关键字段，日期、浏览器类型，点击量
            String dateKey = input.getStringByField("date_key");
            String agent = input.getStringByField("agent");
            int hits = input.getIntegerByField("hits");
            //构建维度组合，日期#浏览器类型，处理计算逻辑实现内容更新
            String key = dateKey + "#" + agent;
            agentSumMap.put(key, agentSumMap.getOrDefault(key, 0L) + hits);
            //发送更新数据
            collector.emit(new Values(key, agentSumMap.get(key)));
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
            throw new RuntimeException("浏览器总点击量聚合失败: " + e.getMessage());
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("agg_key", "sum_hits"));  // agg_key=date#agent
    }
}